/////////////////////////////////////////////////////////////////////////////
// Name:        message_process_impl.h
// Purpose:
// Author:      syy
// Modified by:
// Created:     2015-03-11 17:09:59
// RCS-ID:
// Copyright:
// Licence:
/////////////////////////////////////////////////////////////////////////////
#ifndef _MESSAGE_PROCESS_IMPL_H_
#define _MESSAGE_PROCESS_IMPL_H_

#include <core/container/qbqueue.h>
#include <core/observer/message_observer.h>
#include <core/sync/lock.h>
#include <core/system/event_wrapper.h>
#include <core/system/thread_wrapper.h>
#include <qbprotocol/include/ssCmnDefine.h>

#include <map>
#include <vector>

#include "../../include/message_process.h"
#include "../../include/message_sync.h"

namespace qb {
class CMessageProcessImpl : public IMessageProcess {
 public:
  CMessageProcessImpl();
  virtual ~CMessageProcessImpl();

 public:
  STDMETHOD_(int32_t, Create)(THIS_ INetworkService* ins, IMessageSync* sync);
  STDMETHOD_(int32_t, Destory)(THIS_);
  STDMETHOD_(bool, RegisterMsgProcessorOnThread)
  (THIS_ int, IMessageProcessorOnSubThread* observer);
  STDMETHOD_(void, UnRegisterMsgProcessorOnThread)
  (THIS_ int, IMessageProcessorOnSubThread* observer);
  STDMETHOD_(int32_t, RegisterMessageProcessOnMainThreadObserver)
  (THIS_ IMessageProcessOnMainThreadObserver* observer);
  STDMETHOD_(int32_t, DeRegisterMessageProcessOnMainThreadObserver)
  (THIS_ IMessageProcessOnMainThreadObserver* observer);
  STDMETHOD_(int32_t, PushNetMessage)(THIS_ MESSAGE_SHARED_PTR msp);
  STDMETHOD_(int32_t, PushMainThreadMessage)(THIS_ char* buff, int32_t buflen);

 protected:
  static bool ThreadProcess(void* args);
  bool ThreadProcessMessage(void* args);
  void StopThread(ThreadWrapper** thread_);
  bool ProcessMessage(const char* buff, int buflen);
  bool NotifyMainThreadObserver(int msgtype, char* buff, int buflen);

 private:
  Atomic32 m_exit_flag;
  EventWrapper& m_event;
  ThreadWrapper* m_thread;
  std::map<int, IMessageProcessorOnSubThread*> m_threadMsgProcessors;
  BList<IMessageProcessOnMainThreadObserver*> m_observersMainThread;
  qb::base::Spinlock m_lock;
  threadsafeQueue<MESSAGE_SHARED_PTR> m_queue;
  INetworkService* m_spNetworkService;
  IMessageSync* m_spMessageSync;

  static char s_rcvBuf[Y_RD_SERVICE_MESSAGE_BLOCK_LEN];
  static char s_preBuf[Y_RD_SERVICE_MESSAGE_BLOCK_LEN];
};
}  // namespace qb

#endif  // _MESSAGE_PROCESS_IMPL_H_
